package com.bigdata.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 通过调用DataFrame的API达到查询的效果
 * 使用DSL不需要编写SQL
 * DSL API是需要一定学习成本
 */
object Demo3DSLApi {

  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("dsl")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    spark.sparkContext.setLogLevel("error")
    /**
     * DSL Api
     * JSON 自带schema的，得到的DataFrame就包含了schema信息
     * 不需要自己指定schema
     */
    val studentDF: DataFrame = spark
      .read
      .format("json")
      .load("data/students.json")

    studentDF.printSchema()

    // studentDF.show(false)

    println("-" * 50)

    // 使用列名选择数据
    studentDF
      .select("name", "age", "clazz")
    // .show(false)

    //    // 使用sql表达式
    studentDF
      .selectExpr("name", "age + 1 as age")
    // .show(false)

    // 导入的使用sparkSession这个对象中的隐式转换，这个对象名是什么，这里的spark需要进行替换
    import spark.implicits._
    // 想要在DSL中使用函数，需要导入sparksql中的function
    import org.apache.spark.sql.functions._

    // 使用列对象的方式, 可以直接进行计算
    //使用列对象，可以直接进行sql计算
    studentDF
      .select($"name", $"age" + 1 as "age", $"id")
    //      .show()

    // 在select中使用DSL函数
    // 函数中需要传递column对象，可以使用expr表达式把字符串转为常量的column对象
    studentDF
      .select($"name",
        concat(lit("城管—"), $"name") as "new_name",
        $"age")
    // .show(false)


    studentDF
      .select($"name", $"age"
        , substring($"clazz", 0, 2) as "wlk"
        , concat(substring($"clazz", 0, 2), $"age") as "concat_wlk"
      )
    // .show()

    /**
     * where: 过滤数据
     * 字符串sql表达式, ===表示等于， =!=表示不等于
     */
    studentDF
      .where(substring($"clazz", 0, 2) =!= "文科" and $"age" <= 22)
    // .show()

    /**
     * groupBy agg ： 分组聚合, 分组和聚合要一起使用
     * 分组聚合之后返回的DF只包含分组字段和聚合字段的
     * 能使用的聚合函数和sql中是一样的，count，sum，avg，max，min
     * 查询每个班级的平均年龄
     */
    studentDF
      .where(substring($"clazz", 0, 2) =!= "文科")
      .groupBy($"clazz")
      .agg(avg($"age") as "avg_age1",
        round(avg($"age"), 2) as "avg_age2")
    // .show(false)


    // 按照科目分组聚合
    studentDF
      .groupBy(substring($"clazz", 0, 2) as "subject")
      .agg(round(avg($"age"), 2) as "avg_age")
    // .show()

    /**
     * order by:排序，可以指定升序或倒序
     * 完整的SQL关键字在DSL语法中的使用
     */
    studentDF
      .where($"age" <= 22)
      .groupBy(substring($"clazz", 0, 3) as "sub_clazz")
      .agg(count($"clazz") as "num")
      // 这里的filter是对结果集过滤，类似sql中的having语法
      .filter(substring($"sub_clazz", 0, 2) === "理科")
      .orderBy($"num".desc)
      .limit(3)
    // .show()

    /**
     * join: 表关联
     */
    val scoreDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id string, cid string, sco double")
      .load("data/score.txt")

    //    scoreDF.printSchema()
    //    scoreDF.show(false)

    //关联字段名不一样的写法，关联的是多个字段使用 &&、|| 连接多个条件
    // val joinDF: DataFrame = studentDF.join(scoreDF, $"id" === $"sid" || $"age" === $"cid", "inner")

    // val joinDF: DataFrame = studentDF.join(scoreDF, $"id" === $"sid", "left")
    // 关联字段名称一致, 默认是inner join
    val joinDF: DataFrame = studentDF.join(scoreDF, "id")
    joinDF.show()

    /**
     * 开窗函数
     * 统计每个班级总分前十的学生
     * withColumn: 在DF的基础上增加新的列
     */
    /**
     * 方式1：使用SQL语句的方式
     */
    // 创建一张表，这里的表不是实际意义上的表，是虚拟的
    joinDF.createOrReplaceTempView("t_stu_sco");

    /**
     * 分组去topN
     * 统计每个班级总分前十的学生
     * 通过SQL来查询这个view中的数据
     * 1.得到每个班级每个学生的总分
     */
    val result: DataFrame = spark.sql(
      """
        |select
        |clazz,id,sum_sco,r
        |from (
        |select
        |clazz,id,sum_sco,
        |row_number() over(partition by clazz order by sum_sco desc) as r
        |from (
        |select
        |clazz,id,sum(sco) as sum_sco
        |from
        |t_stu_sco
        | group by clazz,id
        |)
        |) as tmp
        |where tmp.r <= 10
        |""".stripMargin)

    // show 是把结果打印到控制台
    // result.show(100,false)

    // DSL语法实现的窗口函数查询，通过函数得到结果后面按照正确逻辑的函数继续处理即可
    joinDF
      .groupBy($"clazz", $"id")
      .agg(sum($"sco") as "sum_sco")
      // 在原有表的基础上再追加一列
      .withColumn("r", row_number() over(Window.partitionBy($"clazz").orderBy($"sum_sco".desc)))
      .where($"r" <= 10)
      .show(30, false)

  }
}
